package beautiful.butterfly.drds.data_exchange.util;


import beautiful.butterfly.drds.data_exchange.spi.AbstractReader;
import beautiful.butterfly.drds.data_exchange.transport.exchanger.Produce;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLException;
import java.util.List;

public class Reader extends AbstractReader
{


    public static class Job extends AbstractReader.AbstractJob
    {
        private static final Logger LOG = LoggerFactory
                .getLogger(Job.class);

        private Configuration originalConfig = null;
        private beautiful.butterfly.drds.data_exchange.reader.Reader.Job job;


        public void init() throws SQLException
        {
            this.originalConfig = super.getPluginJobConf();

            Integer userConfigedFetchSize = this.originalConfig.getInt(beautiful.butterfly.drds.data_exchange.reader.Constant.FETCH_SIZE);
            if (userConfigedFetchSize != null)
            {
                LOG.warn("对 mysqlreader 不需要配置 fetchSize, mysqlreader 将会忽略这项配置. 如果您不想再看到此警告,请去除fetchSize 配置.");
            }

            this.originalConfig.set(beautiful.butterfly.drds.data_exchange.reader.Constant.FETCH_SIZE, Integer.MIN_VALUE);

            this.job = new beautiful.butterfly.drds.data_exchange.reader.Reader.Job();
            this.job.init(this.originalConfig);
        }

        @Override
        public void preCheck() throws SQLException
        {
            init();
            this.job.preCheck(this.originalConfig);

        }

        @Override
        public List<Configuration> split(int adviceNumber) throws SQLException
        {
            return this.job.split(this.originalConfig, adviceNumber);
        }

        @Override
        public void post()
        {
            this.job.post(this.originalConfig);
        }


        public void destroy()
        {
            this.job.destroy(this.originalConfig);
        }

    }

    public static class Task extends AbstractReader.AbstractTask
    {

        private Configuration readerSliceConfig;
        private beautiful.butterfly.drds.data_exchange.reader.Reader.Task readerTask;


        public void init()
        {
            this.readerSliceConfig = super.getPluginJobConf();
            this.readerTask = new beautiful.butterfly.drds.data_exchange.reader.Reader.Task(super.getTaskGroupId(), super.getTaskId());
            this.readerTask.init(this.readerSliceConfig);

        }


        public void startRead(Produce produce) throws SQLException
        {
            int fetchSize = this.readerSliceConfig.getInt(beautiful.butterfly.drds.data_exchange.reader.Constant.FETCH_SIZE);

            this.readerTask.startRead(this.readerSliceConfig, produce,
                    super.getTaskPluginCollector(), fetchSize);
        }


        public void post()
        {
            this.readerTask.post(this.readerSliceConfig);
        }


        public void destroy()
        {
            this.readerTask.destroy(this.readerSliceConfig);
        }

    }

}
